#include "mqttclient.h"
#include <memory>
#include <string.h>
#include "mosquitto/mosquitto.h"


#define MSGCB_BY_GENERAL_FUNC    0

#if MSGCB_BY_GENERAL_FUNC
/*****************************************************
 * 功能：MQTT消息处理回调函数
 * 输出参数：
 *      pMosInstance：mosquitto实例
 *      pObjCtx：调用的上下文实例
 *      pMessage：MQTT数据消息
 * ***************************************************/
static void pfnOnMqttMessageCallback(struct mosquitto *pMosInstance
                                     , void *pObjCtx
                                     , const struct mosquitto_message *pMessage)
{
    if (!pMosInstance || !pObjCtx || !pMessage) {
        return;
    }

    MqttClient *pObjHandler = static_cast<MqttClient *>(pObjCtx);
    if (pObjHandler) {
        MqttMessageInfo stMqMsgInfo;
        stMqMsgInfo.stTopicInfo.iQOS = pMessage->qos;
        stMqMsgInfo.stTopicInfo.strTopic = std::string(pMessage->topic);
        stMqMsgInfo.strData = std::string(reinterpret_cast<char*>(pMessage->payload), static_cast<size_t>(pMessage->payloadlen));
        pObjHandler->onMessage(stMqMsgInfo);
    }

    return;
}
#endif


/*****************************************************
 * 功能：MQTT客户端lib初始化，全局只调用一次
 * ***************************************************/
void initMqttLibrary()
{
    // libmosquitto 库初始化
    if(MOSQ_ERR_SUCCESS != mosquitto_lib_init()) {
        mosquitto_lib_cleanup();
        printf("[initMqttLibrary] >>> init mosquitto library failed.\r\n");
    }
    return;
}

/*****************************************************
 * 功能：MQTT客户端lib反初始化，全局只调用一次
 * ***************************************************/
void deinitMqttLibrary()
{
    mosquitto_lib_cleanup();
    return;
}

/*****************************************************
 * 功能：构造函数
 * ***************************************************/
MqttClient::MqttClient()
{

}

/*****************************************************
 * 功能：构造函数
 * 输入参数：
 *      stConInfo：连接信息
 * ***************************************************/
MqttClient::MqttClient(const MqttConnectInfo& stConInfo)
{
    memcpy(&m_stConInfo, &stConInfo, sizeof (MqttConnectInfo));
    return;
}

/*****************************************************
 * 功能：析构函数
 * ***************************************************/
MqttClient::~MqttClient()
{
    disconnect();
    memset(&m_stConInfo, 0, sizeof (MqttConnectInfo));
    return;
}

/*****************************************************
 * 功能：设置连接信息
 * 输入参数：
 *      stConInfo：连接信息
 * ***************************************************/
void MqttClient::setConnectInfo(const MqttConnectInfo& stConInfo)
{
    memcpy(&m_stConInfo, &stConInfo, sizeof (MqttConnectInfo));
    return;
}

/*****************************************************
 * 功能：设置回调接口
 * 输入参数：
 *      pContext：调用上下文信息
 *      pfnMsgCb：MQTT消息处理回调函数指针
 * ***************************************************/
void MqttClient::setCallbackFunc(void *pContext, pfnMqttMsgCallback pfnMsgCb)
{
    m_stConInfo.pContext = pContext;
    m_pfnMsgCb = pfnMsgCb;
    return;
}

/*****************************************************
 * 功能：连接MQTT Broker服务
 * 输入参数：
 *      vtTopicInfos：需要订阅的主题信息集合
 *      pfnMsgCb：MQTT消息处理回调函数指针
 * ***************************************************/
void MqttClient::connect(const std::vector<MqttTopicInfo>& vtTopicInfos, pfnMqttMsgCallback pfnMsgCb)
{
    if (m_bConnectflag) {
        printf("[MqttClient::connect] >>> client is already connected to mqtt Broker, please disconnect first.\r\n");
        return;
    }

    // 如果mosquitto非空，则表示在等待网络返回，无需再调用连接
    if (m_pMqttClient) {
        printf("[MqttClient::connect] >>> please wait for network response, or call disconnect function.\r\n");
        return;
    }

    m_pMqttClient = mosquitto_new(nullptr, true, this);
    if(!m_pMqttClient) {
        printf("[MqttClient::connect] >>> new mosquitto instance failed, please call disconnect and deinitMqttLibrary function.\r\n");
        return;
    }

    if (m_mt4TopicInfos.try_lock()) {
        m_vtTopicInfos = vtTopicInfos;
        //m_vtTopicInfos = std::move(vtTopicInfos); // move后原来的清空了，慎用
        m_mt4TopicInfos.unlock();
    }

    // 设置数据回调处理函数
#if MSGCB_BY_GENERAL_FUNC
    // 常见方式
    mosquitto_message_callback_set(m_pMqttClient, pfnOnMqttMessageCallback);
#else
    // lamda函数
    mosquitto_message_callback_set(m_pMqttClient, [](struct mosquitto *pMosInstance, void *pObjCtx
                                   , const struct mosquitto_message *pMessage) {
        if (!pMosInstance || !pObjCtx || !pMessage) {
            return;
        }

        MqttClient *pObjHandler = static_cast<MqttClient *>(pObjCtx);
        if (pObjHandler) {
            MqttMessageInfo stMqMsgInfo;
            stMqMsgInfo.stTopicInfo.iQOS = pMessage->qos;
            stMqMsgInfo.stTopicInfo.strTopic = std::string(pMessage->topic);
            stMqMsgInfo.strData = std::string(reinterpret_cast<char*>(pMessage->payload), static_cast<size_t>(pMessage->payloadlen));
            pObjHandler->onMessage(stMqMsgInfo);
        }

        return;
    });
#endif

    // 设置连接处理函数
    // 使用std::shared_ptr管理对象生命周期
    //auto self = std::shared_ptr<PML::MqttClient>(this, [](PML::MqttClient*) {});
    //mosquitto_connect_callback_set(m_pMqttClient, [self](struct mosquitto *pMosInstance, void *pObjCtx, int iRet) {
    mosquitto_connect_callback_set(m_pMqttClient, [](struct mosquitto *pMosInstance, void *pObjCtx, int iRet) {
        (void)pMosInstance;
        MqttClient *pObjHandler = static_cast<MqttClient *>(pObjCtx);
        if (pObjHandler) {
            pObjHandler->onConnect(iRet);
        }
    });

    if (!(m_stConInfo.bAnonymous)) {
        if (MOSQ_ERR_SUCCESS != mosquitto_username_pw_set(m_pMqttClient, m_stConInfo.strAccount.data()
                                                          , m_stConInfo.strPwd.data())) {
            printf("[MqttClient::connect] >>> set mosquitto client account and password failed, it will be disconnect later.\r\n");
            disconnect();
            return;
        }
    }

    int iRet = mosquitto_connect(m_pMqttClient, m_stConInfo.strBrokerIp.data()
                                 , m_stConInfo.ui16BrokerPort, m_stConInfo.ui16KeepAliveInterval);
    if(MOSQ_ERR_SUCCESS != iRet) {
        printf("[MqttClient::connect] >>> connect to mqtt broker ret: %d, info: %s.\r\n", iRet, mosquitto_strerror(iRet));
        return;
    }

    m_pfnMsgCb = pfnMsgCb;
    mosquitto_loop_start(m_pMqttClient);

    return;
}

/*****************************************************
 * 功能：断开与MQTT Broker服务的连接
 * ***************************************************/
void MqttClient::disconnect()
{
    m_pfnMsgCb = nullptr;
    if (m_pMqttClient) {
        mosquitto_loop_stop(m_pMqttClient, true);
        mosquitto_disconnect(m_pMqttClient);
        mosquitto_destroy(m_pMqttClient);
        m_pMqttClient = nullptr;
    }

    if (m_mt4TopicInfos.try_lock()) {
        m_vtTopicInfos.clear();
        m_mt4TopicInfos.unlock();
    }

    m_bConnectflag = false;
    m_iMsgID.store(0);

    return;
}

/*****************************************************
 * 功能：获取是否与MQTT Broker已连接
 * 返回值：
 *      bool：连接结果，true - 已连接，false - 未连接
 * ***************************************************/
bool MqttClient::isConnected() const
{
    return m_bConnectflag;
}

/*****************************************************
 * 功能：订阅数据消息，连接成功后调用
 * 输入参数：
 *      strTopic：主题
 *      iQOS：质量
 * ***************************************************/
void MqttClient::subscribe(const std::string& strTopic, int iQOS)
{
    if (!queryTopicInfo(strTopic)) {
        MqttTopicInfo stTopicInfo;
        stTopicInfo.strTopic = strTopic;
        stTopicInfo.iQOS = iQOS;
        subscribeAfterConnected(stTopicInfo);

        // 添加到集合中
        if (m_mt4TopicInfos.try_lock()) {
            m_vtTopicInfos.emplace_back(stTopicInfo);
            m_mt4TopicInfos.unlock();
        }
    } else {
        printf("[MqttClient::subscribe] >>> topic %s has been subscribed, it can not be subscribed again.\r\n"
               , strTopic.data());
    }

    return;
}

/*****************************************************
 * 功能：取消订阅数据消息
 * 输入参数：
 *      strTopic：主题
 * ***************************************************/
void MqttClient::unsubscribe(const std::string& strTopic)
{
    if (strTopic.empty()) {
        return;
    }

    if (m_pMqttClient) {
        mosquitto_unsubscribe(m_pMqttClient, nullptr, strTopic.data());
    }

    // 从主题集合中移出
    if (m_mt4TopicInfos.try_lock()) {
        for (auto iter = m_vtTopicInfos.begin(); iter != m_vtTopicInfos.end(); ++iter) {
            if (0 == iter->strTopic.compare(strTopic.data())) {
                iter = m_vtTopicInfos.erase(iter);
                break;
            }
        }
        m_mt4TopicInfos.unlock();
    }

    return;
}

/*****************************************************
 * 功能：取消订阅数据消息
 * 输入参数：
 *      vtTopics：主题集合
 * ***************************************************/
void MqttClient::unsubscribe(const std::vector<std::string>& vtTopics)
{
    for (size_t i = 0, size = vtTopics.size(); i < size; ++i) {
        unsubscribe(vtTopics[i]);
    }

    return;
}

/*****************************************************
 * 功能：发布数据消息
 * 输入参数：
 *      stMqMsgInfo：MQTT数据消息
 * ***************************************************/
void MqttClient::publish(const MqttMessageInfo& stMqMsgInfo)
{
    if (stMqMsgInfo.strData.empty() || stMqMsgInfo.stTopicInfo.strTopic.empty()) {
        return;
    }

    if (m_bConnectflag && m_pMqttClient) {
        int iMsgId = ++m_iMsgID;
        int iRet = mosquitto_publish(m_pMqttClient, &(iMsgId), stMqMsgInfo.stTopicInfo.strTopic.data()
                                     , static_cast<int>(stMqMsgInfo.strData.size()), stMqMsgInfo.strData.data()
                                     , stMqMsgInfo.stTopicInfo.iQOS, false);
        if (MOSQ_ERR_SUCCESS != iRet) {
            printf("[MqttClient::publish] >>> publish topic: %s failed, ret: %d, info: %s.\r\n"
                   , stMqMsgInfo.stTopicInfo.strTopic.data(), iRet, mosquitto_strerror(iRet));
        }
    }
    return;
}

/*****************************************************
 * 功能：处理连接状态，连接回调函数中调用
 * 输入参数：
 *      iRetCode：MQTT连接处理返回值
 * ***************************************************/
void MqttClient::onConnect(int iRetCode)
{
    if (MOSQ_ERR_SUCCESS == iRetCode) {
        m_bConnectflag = true;
        m_iMsgID.store(0);

        // 订阅消息
        std::vector<MqttTopicInfo> vtTopicInfos;
        if (m_mt4TopicInfos.try_lock()) {
            vtTopicInfos = m_vtTopicInfos;
            m_mt4TopicInfos.unlock();
        }

        for (size_t i = 0, size = vtTopicInfos.size(); i < size; ++i) {
            subscribeAfterConnected(vtTopicInfos[i]);
        }
        printf("[MqttClient::onConnect] >>> connected to %s.\r\n", m_stConInfo.strBrokerIp.data());
    } else {
        printf("[MqttClient::onConnect] >>> connect to mqtt Broker ret: %d, info: %s.\r\n", iRetCode, mosquitto_strerror(iRetCode));
        m_bConnectflag = false;
        m_iMsgID.store(0);
    }

    return;
}

/*****************************************************
 * 功能：处理数据消息
 * 输入参数：
 *      stMqMsgInfo：MQTT数据消息
 * ***************************************************/
void MqttClient::onMessage(const MqttMessageInfo& stMqMsgInfo)
{
    if (m_pfnMsgCb) {
        // 只有订阅了该主题才进行数据回调
        if (queryTopicInfo(stMqMsgInfo.stTopicInfo.strTopic)) {
            m_pfnMsgCb(m_stConInfo.pContext, stMqMsgInfo);
        } else {
            printf("[MqttClient::onMessage] >>> topic %s is not be subscribed, it can not be call back to deal.\r\n"
                   , stMqMsgInfo.stTopicInfo.strTopic.data());
        }
    }
    return;
}

/*****************************************************
 * 功能：订阅数据消息，连接成功后调用
 * 输入参数：
 *      stTopicInfo：主题信息
 * ***************************************************/
void MqttClient::subscribeAfterConnected(const MqttTopicInfo& stTopicInfo)
{
    if (stTopicInfo.strTopic.empty()) {
        return;
    }

    if (m_bConnectflag && m_pMqttClient)
    {
        int iRet = mosquitto_subscribe(m_pMqttClient, nullptr, stTopicInfo.strTopic.data(), stTopicInfo.iQOS);
        if (MOSQ_ERR_SUCCESS != iRet) {
            printf("[MqttClient::subscribeAfterConnected] >>> subscribe topic: %s failed, ret: %d, info: %s.\r\n"
                   , stTopicInfo.strTopic.data(), iRet, mosquitto_strerror(iRet));
        } else {
            printf("[MqttClient::subscribeAfterConnected] >>> subscribe topic: %s succeed.\r\n"
                   , stTopicInfo.strTopic.data());
        }
    }

    return;
}

/*****************************************************
 * 功能：查找是否已经订阅过该主题了
 * 输入参数：
 *      strTopic：主题信息
 * 返回值：
 *      bool：查找结果，true - 已存在，false - 未存在
 * ***************************************************/
bool MqttClient::queryTopicInfo(const std::string& strTopic)
{
    bool bRet = false;
    std::vector<MqttTopicInfo> vtTopicInfos;
    if (m_mt4TopicInfos.try_lock()) {
        vtTopicInfos = m_vtTopicInfos;
        m_mt4TopicInfos.unlock();
    }

    for (size_t i = 0, size = vtTopicInfos.size(); i < size; ++i) {
        if (0 == vtTopicInfos[i].strTopic.compare(strTopic.data())) {
            bRet = true;
            break;
        }
    }

    return bRet;
}

